package com.atliuzu.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Properties;

/**
 * @Author : songyuan
 * @Description :
 * Date : 2022/7/18 22:05
 * @Version : 1.0
 */
public class MyKafkaUtil {

    private static Properties prop = new Properties();
    private static final String KAFKA_SERVER = "hadoop03:9092";
    static {
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
    }
    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic, String groupId) {

        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        return new FlinkKafkaConsumer<String>(topic, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String s) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
                if (consumerRecord == null || consumerRecord.value() == null) {
                    return null;
                } else {

                    return new String(consumerRecord.value());
                }
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return TypeInformation.of(String.class);
            }
        }, prop);
    }
    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic) {
        return new FlinkKafkaProducer<String>(topic,
                new SimpleStringSchema(),
                prop);
    }

    //DDL格式 消费Kafka topic_db数据
    public static String getKafkaDDL(String topic, String groupId) {
        return " WITH ( " +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                "  'properties.group.id' = '" + groupId + "', " +
                "  'scan.startup.mode' = 'latest-offset', " +
                "  'format' = 'json' " +
                ")";
    }
    //从maxwell 中读取数据  并建表（实际生产，时间可从data中获取）
    public static String getTopicDb(String groupId) {
        return "CREATE TABLE topic_db ( " +
                "  `database` STRING, " +
                "  `table` STRING, " +
                "  `type` STRING, " +
                "  `ts` BIGINT, " +
                "  `data` Map<STRING,STRING>, " +
                "  `old` Map<STRING,STRING>, " +
                "  `pt` AS PROCTIME() " +
                ")" + getKafkaDDL("topic_db", groupId);
    }

    public static String getInsertKafkaDDL(String topic) {
        return " WITH ( " +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                "  'format' = 'json' " +
                ")";
    }

    public static String getUpsertKafkaDDL(String topic) {
return "" +
        " WITH ( " +
        "  'connector' = 'upsert-kafka', " +
        "  'topic' = '"+topic+"', " +
        "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
        "  'key.format' = 'json', " +
        "  'value.format' = 'json' " +
        ")";


    }
}
